package title2

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import title1.IPUtils

/**
  * 问题2.计算每个省城市成交量的top3（结果保存到MySQL中）
  * SparkCore做法
  */
object Top3ofProvincialCities {
  def main(args: Array[String]): Unit = {
    //1 处理全国ip数据并得到匹配规则

    val conf = new SparkConf().setMaster("local[*]").setAppName(this.getClass.getName)
    val sc = new SparkContext(conf)

    //匹配规则数据源
    val ipRulesData: RDD[String] = sc.textFile("data/ip.txt")
    //把起始位置，终点位置，省份,城市返回
    val ipRulesRDD: RDD[(Long, Long, String, String)] = ipRulesData.map({
      t =>
        val rules: Array[String] = t.split("[|]")
        val start = rules(2).toLong
        val end = rules(3).toLong
        val province = rules(6)
        val city = rules(7)
        (start, end, province, city)
    })

    val ipRules = ipRulesRDD.collect()
    //将ip匹配规则广播出去
    val ipRuleBC: Broadcast[Array[(Long, Long, String, String)]] = sc.broadcast(ipRules)

    //2 处理订单数据
    val orderData = sc.textFile("data/订单数据.log")
    val provinceCityAndAmount: RDD[((String, String), Int)] = orderData.map({ t =>
      val line: Array[String] = t.split(" ")
      //2.1获取ip
      val ip = line(1)
      val ipnum: Long = IPUtils.ip2Long(ip)
      //2.2获取消费额
      val amount = line(line.length - 1).toInt
      //2.3从广播变量中获取匹配规则
      val ipRulesFromBC: Array[(Long, Long, String, String)] = ipRuleBC.value
      //2.4根据规则获取ip对应省
      val index: Int = IPUtils.searchCity(ipnum, ipRulesFromBC)
      if (index != -1) {
        val province = ipRulesFromBC(index)._3
        val city = ipRulesFromBC(index)._4
        ((province, city), amount)
      } else {
        (("unknown", "unknown"), 0)
      }
    })


    //统计每个省的每个城市的成交额
    val CityAmount: RDD[((String, String), Int)] = provinceCityAndAmount.reduceByKey(_ + _)
    //根据省分组，在每组中以每个城市的成交额排序  取前3
    val result: RDD[List[((String, String), Int)]] = CityAmount.groupBy(_._1._1).map({ t =>
      t._2.toList.sortBy(-_._2).take(3)
    })
    result.foreachPartition({t=>
      data2Mysql(t,"Top3ofProvincialCities")
    })

  }

  //数据写入mysql函数
  val data2Mysql = (it: Iterator[List[((String,String) ,Int)]], tableName: String) => {
    // 使用foreachPartition  迭代的变量是一个分区的数据，迭代器
    var conn: Connection = null
    var pstm: PreparedStatement = null
    // try  catch
    try {
      // 执行数据库的操作    如果在集群中运行，mysql的ip地址不能写成localhost，具体的ip
      val url = "jdbc:mysql://192.168.8.1:3306/scott?characterEncoding=utf-8"
      // 获取连接
      conn = DriverManager.getConnection(url, "root", "123456")
      val sql = "insert into " + tableName + " values(?,?,?)"
      pstm = conn.prepareStatement(sql)
      // 因为函数的参数，是一个迭代器类型，而不是具体的元素值，所以 需要多迭代器进行循环，在循环中，进行业务逻辑的操作
      it.foreach({
        t =>t.map({t=>
          pstm.setString(1, t._1._1)
          pstm.setString(2, t._1._2)
          pstm.setInt(3, t._2)
          pstm.executeUpdate()
        })
      })
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      // 关闭数据库的连接
      if (pstm != null) pstm.close()
      if (conn != null) conn.close()
    }
  }
}
